package rocketmq

import (
	"context"
	"fmt"
	//"github.com/apache/rocketmq-clients/golang/credentials"
	//"github.com/apache/rocketmq-clients/golang"
	"github.com/apache/rocketmq-clients/golang/v5"
	"github.com/apache/rocketmq-clients/golang/v5/credentials"
	"log"
	"os"
	"strconv"
	"time"
)

const (
	TopicTrans = "test_trasaction"
)

// 事物消息生产者
func TransProducer() {
	os.Setenv("mq.consoleAppender.enabled", "true")
	golang.ResetLogger()
	// new producer instance
	producer, err := golang.NewProducer(&golang.Config{
		Endpoint: Endpoint,
		//NameSpace: "",
		Credentials: &credentials.SessionCredentials{
			AccessKey:    AccessKey,
			AccessSecret: SecretKey,
		},
	},
		golang.WithTopics(TopicTrans),

		// 构造事务生产者：事务消息需要生产者构建一个事务检查器，用于检查确认异常半事务的中间状态。
		golang.WithTransactionChecker(&golang.TransactionChecker{
			Check: func(msg *golang.MessageView) golang.TransactionResolution {
				log.Printf("check transaction message=%v pro=%v", msg, msg.GetProperties())

				return golang.COMMIT
			},
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	// start producer
	err = producer.Start()
	if err != nil {
		log.Fatal(err)
	}
	// gracefule stop producer
	defer producer.GracefulStop()

	for i := 0; i < 1; i++ {
		// new a message
		msg := &golang.Message{
			Topic: TopicTrans,
			Body:  []byte("this is a message : " + strconv.Itoa(i)),
		}
		// set keys and tag
		msg.SetKeys("a", "b")
		msg.SetTag("ab")
		//一般事务消息都会设置一个本地事务关联的唯一ID，用来做本地事务回查的校验。
		msg.AddProperty("orderId", "1")

		// send message in sync
		transaction := producer.BeginTransaction()
		resp, err := producer.SendWithTransaction(context.TODO(), msg, transaction)
		if err != nil {
			log.Fatal(err)
		}

		for i := 0; i < len(resp); i++ {
			fmt.Printf("xxxxxx%#v\n", resp[i])
		}

		//todo local transaction
		var localSuccess bool

		if localSuccess {
			// commit transaction message
			err = transaction.Commit()
			if err != nil {
				log.Fatal(err)
			}
		} else {
			transaction.RollBack()
		}

		// wait a moment
		time.Sleep(time.Second * 1)

	}
}
